package com.shujia.core

import java.sql.PreparedStatement

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcSink, JdbcStatementBuilder}
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.util.Collector

object Demo3Car {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 开启flink的快照
      *
      */
    env.enableCheckpointing(1000)
    /**
      * 1、从kafka中读取过车数据
      *
      */

    val source: KafkaSource[String] = KafkaSource
      .builder[String]
      .setBootstrapServers("master:9092")
      .setTopics("cars")
      .setGroupId("my-group")
      .setStartingOffsets(OffsetsInitializer.earliest)
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build

    val catsDS: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")

    /**
      * 1、统计每个城市总的车流量
      * 2、统计每个道路的车流量
      *
      */

    val cityTag: OutputTag[(String, Int)] = OutputTag[(String, Int)]("city")
    val roadTag: OutputTag[(Long, Int)] = OutputTag[(Long, Int)]("road")

    val sideOutDS: DataStream[String] = catsDS.process(new ProcessFunction[String, String] {

      override def processElement(car: String,
                                  ctx: ProcessFunction[String, String]#Context,
                                  out: Collector[String]): Unit = {


        /**
          * 使用fastJson解析json数据
          *
          */

        val carJson: JSONObject = JSON.parseObject(car)
        //取出成都编号和道路编号
        val cityCode: String = carJson.getString("city_code")
        val roadDd: Long = carJson.getLong("road_id")


        //将数据发生到下游
        ctx.output(cityTag, (cityCode, 1))
        ctx.output(roadTag, (roadDd, 1))
      }
    })


    //取出城市的数据
    val cityDS: DataStream[(String, Int)] = sideOutDS.getSideOutput(cityTag)
    //统计城市的车流量
    val cityFlow: DataStream[(String, Int)] = cityDS.keyBy(_._1).sum(1)

    /**
      * 将数据保存到mysql
      *
      */

    val citySink: SinkFunction[(String, Int)] = JdbcSink.sink(
      "replace into city_flow (city_id,flow) values(?,?)",
      new JdbcStatementBuilder[(String, Int)] {
        override def accept(stat: PreparedStatement, kv: (String, Int)): Unit = {
          stat.setString(1, kv._1)
          stat.setInt(2, kv._2)
        }
      },
      //jdbc链接参数
      new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8")
        .withDriverName("com.mysql.jdbc.Driver")
        .withUsername("root")
        .withPassword("123456")
        .build()
    )

    cityFlow.addSink(citySink)


    //取出道路的数据
    val roadDS: DataStream[(Long, Int)] = sideOutDS.getSideOutput(roadTag)
    //统计道路的车流量
    val roadFlow: DataStream[(Long, Int)] = roadDS.keyBy(_._1).sum(1)


    /**
      * JdbcSink: 将数据写入到mysql的时候默认是5000条数据刷新一次'
      * 当开启checkpoint之后也会刷新
      *
      * replace into : 如果不存在就插入,如果存在就替换
      *
      */
    val roadSink: SinkFunction[(Long, Int)] = JdbcSink.sink(
      "replace into road_flow (road_id,flow) values(?,?)",
      new JdbcStatementBuilder[(Long, Int)] {
        override def accept(stat: PreparedStatement, kv: (Long, Int)): Unit = {
          stat.setLong(1, kv._1)
          stat.setInt(2, kv._2)
        }
      },
      //jdbc链接参数
      new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8")
        .withDriverName("com.mysql.jdbc.Driver")
        .withUsername("root")
        .withPassword("123456")
        .build()
    )

    roadFlow.addSink(roadSink)

    env.execute()


  }

}
